#include <chrono>
#include <thread>  
#include <future>
#include <iostream>
#include "../corelib/network/network.hh"
#include "unittest/unittest.hh"

using namespace kratos::network;

static bool test_accept(const std::string& iotype) {
    bool result = false;
    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345");
    acceptor.onAccept([&](kratos::network::Session& session) {
        result = true;
    });
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype).connect("127.0.0.1:12345");
    connector.start();
    std::thread t([&](){
        while (!result) {
            acceptor.update();
            connector.update();
        }
    });
    bool value = false;
    CHECK_WAIT_TIMEOUT(result, 5000);
    value = result;
    result = true;
    t.join();
    acceptor.stopWait();
    connector.stopWait();
    return value;
}

static bool test_connect(const std::string& iotype) {
    bool result = false;
    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345");
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype);
    connector.connect("127.0.0.1:12345").onConnect([&](kratos::network::Session& session) {
        result = true;
    });
    connector.start();
    bool run = true;
    std::thread t([&]() {
        while (run) {            
            acceptor.update();
            connector.update();
        }
    });
    CHECK_WAIT_TIMEOUT(result, 5000);
    run = false;
    t.join();
    acceptor.closeAll();
    connector.closeAll();
    acceptor.update();
    connector.update();
    acceptor.stopWait();
    connector.stopWait();
    return result;
}

static bool test_acceptor_session_read(const std::string& iotype) {
    bool result = false;
    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345").onReceive([&](kratos::network::Session& session) {
        result = true;
    });
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype);
    connector.connect("127.0.0.1:12345").onConnect([&](kratos::network::Session& session) {
        int i = 0;
        session.getWriter().talk(i);
    });
    connector.start();
    bool run = true;
    std::thread t([&]() {
        while (run) {            
            acceptor.update();
            connector.update();
        }
    });
    CHECK_WAIT_TIMEOUT(result, 5000);
    run = false;
    t.join();
    acceptor.stopWait();
    connector.stopWait();
    return result;
}

static bool test_connector_session_read(const std::string& iotype) {
    bool result = false;
    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345").onAccept([&](kratos::network::Session& session) {
        int i = 0;
        session.getWriter().talk(i);
    });
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype);
    connector.connect("127.0.0.1:12345").onReceive([&](kratos::network::Session& session) {
        result = true;
    });
    connector.start();
    bool run = true;
    std::thread t([&]() {
        while (run) {            
            acceptor.update();
            connector.update();
        }
    });
    CHECK_WAIT_TIMEOUT(result, 5000);
    run = false;
    t.join();
    acceptor.closeAll();
    connector.closeAll();
    acceptor.update();
    connector.update();
    acceptor.stopWait();
    connector.stopWait();
    return result;
}

#include <unordered_set>
static bool test_session_ID(const std::string& iotype) {
    constexpr int SESSION_COUNT = 10;
    bool result = false;
    std::unordered_set<std::uint32_t> sessions;
    bool return_value = true;
    int session_count = 0;
    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345").onAccept([&](kratos::network::Session& session) {
        if (sessions.find(session.getID()) != sessions.end()) {
            return_value = false;
        }
        sessions.insert(session.getID());
        session_count++;
        if (session_count == SESSION_COUNT) {
            result = true;
        }
    });
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype);
    for (int i = 0; i < SESSION_COUNT; i++) {
        connector.connect("127.0.0.1:12345");
    }
    connector.start();
    bool run = true;
    std::thread t([&]() {
        while (run) {
            acceptor.update();
            connector.update();
        }
    });
    CHECK_WAIT_TIMEOUT(session_count == SESSION_COUNT, 5000);
    run = false;
    t.join();
    acceptor.closeAll();
    connector.closeAll();
    acceptor.update();
    connector.update();
    acceptor.stopWait();
    connector.stopWait();
    return return_value && (session_count == SESSION_COUNT);
}

static bool test_session_read(const std::string& iotype) {
    bool fail = false;
    int last = 0;
    constexpr int MAX_VALUE = 1000;
    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345").onReceive([&](kratos::network::Session& session) {
        auto& reader = session.getReader();
        for(;reader.available() && (reader.available() >= sizeof(int));) {
            int c = 0;
            reader.read(c);
            if (c != last + 1) {                
                fail = true;
            }
            last++;
        }
    });
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype);
    connector.connect("127.0.0.1:12345").onConnect([&](kratos::network::Session& session) {
        for (int i = 1; i <= MAX_VALUE; i++) {
            session.getWriter().talk(i);
        }
    });
    connector.start();
    bool run = true;
    std::thread t([&]() {
        while (run) {
            acceptor.update();
            connector.update();
        }
    });
    CHECK_WAIT_TIMEOUT(last == MAX_VALUE, 5000);
    run = false;
    t.join();
    acceptor.closeAll();
    connector.closeAll();
    acceptor.update();
    connector.update();
    acceptor.stopWait();
    connector.stopWait();
    return (!fail) && (last == MAX_VALUE);
}

static bool test_multi_session_read(const std::string& iotype) {
    bool fail = false;
    int sendValue = 1;
    int last = 0;
    constexpr int MAX_VALUE = 10;
    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345").onReceive([&](kratos::network::Session& session) {
        auto& reader = session.getReader();
        for (; reader.available() && (reader.available() >= sizeof(int));) {
            int c = 0;
            reader.read(c);
            if (c != last + 1) {
                fail = true;
            }
            last++;
        }
    });
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype);
    for (int i = 1; i <= MAX_VALUE; i++) {
        connector.connect("127.0.0.1:12345").onConnect([&](kratos::network::Session& session) {
            session.getWriter().talk(sendValue);
            sendValue++;
        });
    }
    connector.start();
    bool run = true;
    std::thread t([&]() {
        while (run) {
            acceptor.update();
            connector.update();
        }
    });
    CHECK_WAIT_TIMEOUT(last == MAX_VALUE, 5000);
    run = false;
    t.join();
    acceptor.stopWait();
    connector.stopWait();
    return (!fail) && (last == MAX_VALUE);
}

FIXTURE_BEGIN(Network)

CASE(SessionAccept) {
    ASSERT_TRUE(test_accept("tcp"));
    ASSERT_TRUE(test_accept("udp"));
    ASSERT_TRUE(test_accept("enet"));
    ASSERT_TRUE(test_accept("kcp"));
}

CASE(SessionConnect) {
    ASSERT_TRUE(test_connect("tcp"));
    ASSERT_TRUE(test_connect("udp"));
    ASSERT_TRUE(test_connect("enet"));
    ASSERT_TRUE(test_connect("kcp"));
}

CASE(AcceptorSessionRead) {
    ASSERT_TRUE(test_acceptor_session_read("tcp"));
    ASSERT_TRUE(test_acceptor_session_read("udp"));
    ASSERT_TRUE(test_acceptor_session_read("enet"));
    ASSERT_TRUE(test_acceptor_session_read("kcp"));
}

CASE(ConnectorSessionRead) {
    ASSERT_TRUE(test_connector_session_read("tcp"));
    ASSERT_TRUE(test_connector_session_read("udp"));
    ASSERT_TRUE(test_connector_session_read("enet"));
    ASSERT_TRUE(test_connector_session_read("kcp"));
}

CASE(AcceptorSessionID) {
    ASSERT_TRUE(test_session_ID("tcp"));
    ASSERT_TRUE(test_session_ID("udp"));
    ASSERT_TRUE(test_session_ID("enet"));
    ASSERT_TRUE(test_session_ID("kcp"));
}

CASE(SessionRead) {
    ASSERT_TRUE(test_session_read("tcp"));
    ASSERT_TRUE(test_session_read("udp"));
    ASSERT_TRUE(test_session_read("enet"));
    ASSERT_TRUE(test_session_read("kcp"));
}

CASE(MultiSessionRead) {
    ASSERT_TRUE(test_multi_session_read("tcp"));
    ASSERT_TRUE(test_multi_session_read("udp"));
    ASSERT_TRUE(test_multi_session_read("enet"));
    ASSERT_TRUE(test_multi_session_read("kcp"));
}

static bool test_acceptor_multi_start(const std::string& iotype) {
    auto& acceptor = defaultAcceptor(iotype);
    acceptor.start();
    return acceptor.start();
}

CASE(AcceptorMultiStart) {
    ASSERT_TRUE(test_acceptor_multi_start("tcp"));
    ASSERT_TRUE(test_acceptor_multi_start("udp"));
    ASSERT_TRUE(test_acceptor_multi_start("enet"));
    ASSERT_TRUE(test_acceptor_multi_start("kcp"));
}

static bool test_connector_multi_start(const std::string& iotype) {
    auto& connector = defaultConnector(iotype);
    connector.start();
    return connector.start();
}

CASE(ConnectorMultiStart) {
    ASSERT_TRUE(test_connector_multi_start("tcp"));
    ASSERT_TRUE(test_connector_multi_start("udp"));
    ASSERT_TRUE(test_connector_multi_start("enet"));
    ASSERT_TRUE(test_connector_multi_start("kcp"));
}

static bool test_acceptor_stop_but_not_start(const std::string& iotype) {
    auto& acceptor = defaultAcceptor(iotype);
    acceptor.stopWait();
    return true;
}

CASE(AcceptorStopButNotStart) {
    ASSERT_NOEXCEPT(test_acceptor_stop_but_not_start("tcp"));
    ASSERT_NOEXCEPT(test_acceptor_stop_but_not_start("udp"));
    ASSERT_NOEXCEPT(test_acceptor_stop_but_not_start("enet"));
    ASSERT_NOEXCEPT(test_acceptor_stop_but_not_start("kcp"));
}

static bool test_connector_stop_but_not_start(const std::string& iotype) {
    auto& connector = defaultConnector(iotype);
    connector.stopWait();
    return true;
}

CASE(ConnectorStopButNotStart) {
    ASSERT_NOEXCEPT(test_connector_stop_but_not_start("tcp"));
    ASSERT_NOEXCEPT(test_connector_stop_but_not_start("udp"));
    ASSERT_NOEXCEPT(test_connector_stop_but_not_start("enet"));
    ASSERT_NOEXCEPT(test_connector_stop_but_not_start("kcp"));
}

static bool test_acceptor_multi_stop(const std::string& iotype) {
    auto& selector = defaultAcceptor(iotype);
    selector.start();
    selector.stopWait();
    selector.stopWait();
    return true;
}

CASE(AcceptorMultiStop) {
    ASSERT_NOEXCEPT(test_acceptor_multi_stop("tcp"));
    ASSERT_NOEXCEPT(test_acceptor_multi_stop("udp"));
    ASSERT_NOEXCEPT(test_acceptor_multi_stop("enet"));
    ASSERT_NOEXCEPT(test_acceptor_multi_stop("kcp"));
}

static bool test_connector_multi_stop(const std::string& iotype) {
    auto& selector = defaultConnector(iotype);
    selector.start();
    selector.stopWait();
    selector.stopWait();
    return true;
}

CASE(ConnectorMultiStop) {
    ASSERT_NOEXCEPT(test_connector_multi_stop("tcp"));
    ASSERT_NOEXCEPT(test_connector_multi_stop("udp"));
    ASSERT_NOEXCEPT(test_connector_multi_stop("enet"));
    ASSERT_NOEXCEPT(test_connector_multi_stop("kcp"));
}

static bool test_accept_session_close(const std::string& iotype) {
    bool result = false;
    bool connected = false;
    bool accepted = false;
    std::uint32_t id = 0;
    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345").onAccept([&](kratos::network::Session& session) {
        id = session.getID();
        accepted = true;
    }).onClose([&](kratos::network::Session& session) {
        result = true;
    });
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype).connect("127.0.0.1:12345").onConnect([&](kratos::network::Session& session) {
        connected = true;
    });
    connector.start();
    std::future<void> future = std::async(std::launch::async, [&]() {
        while (!connected || !accepted) {
            acceptor.update();
            connector.update();
        }
    });
    future.wait();
    acceptor.close(id);
    while (!result) {
        acceptor.update();
        connector.update();
    }
    acceptor.stopWait();
    connector.stopWait();
    return result;
}

CASE(AcceptSessionClose) {
    ASSERT_TRUE(test_accept_session_close("tcp"));
    ASSERT_TRUE(test_accept_session_close("udp"));
    ASSERT_TRUE(test_accept_session_close("enet"));
    ASSERT_TRUE(test_accept_session_close("kcp"));
}

static bool test_connect_session_close(const std::string& iotype) {
    bool result = false;
    bool connected = false;
    bool accepted = false;
    std::uint32_t id = 0;

    auto& acceptor = defaultAcceptor(iotype).listen("127.0.0.1:12345").onAccept([&](kratos::network::Session& session) {
        accepted = true;
    });
    acceptor.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    auto& connector = defaultConnector(iotype).connect("127.0.0.1:12345").onConnect([&](kratos::network::Session& session) {
        id = session.getID();
        connected = true;
    }).onClose([&](kratos::network::Session& session) {
        result = true;
    });;
    connector.start();
    std::future<void> future = std::async(std::launch::async, [&]() {
        while (!connected || !accepted) {
            acceptor.update();
            connector.update();
        }
    });
    future.wait();
    connector.close(id);
    while (!result) {
        acceptor.update();
        connector.update();
    }
    acceptor.stopWait();
    connector.stopWait();
    return result;
}

CASE(ConnectSessionClose) {
    ASSERT_TRUE(test_connect_session_close("tcp"));
    ASSERT_TRUE(test_connect_session_close("udp"));
    ASSERT_TRUE(test_connect_session_close("enet"));
    ASSERT_TRUE(test_connect_session_close("kcp"));
}

FIXTURE_END(Network)
